package com.bridgeintelligent.tag.bulkload.service.core;

import com.bridgeintelligent.tag.bulkload.api.LookLikeExportDto;
import com.bridgeintelligent.tag.bulkload.api.fileload.Filter;
import com.bridgeintelligent.tag.bulkload.api.fileload.FilterGroup;
import com.bridgeintelligent.tag.bulkload.api.issue.IssueLog;
import com.bridgeintelligent.tag.bulkload.api.issue.IssueResult;
import com.bridgeintelligent.tag.bulkload.api.issue.IssueTaskLog;
import com.bridgeintelligent.tag.bulkload.service.core.fileload.ConditionBuilder;
import com.bridgeintelligent.tag.constants.bulkload.IssueMqMsg;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.mapper.EsTimeMapper;
import com.bridgeintelligent.tag.mapper.FileExportMapper;
import com.bridgeintelligent.tag.mapper.IssueTaskMapper;
import com.bridgeintelligent.tag.utils.FieldHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;

/**
 * @description
 * @author: grp 2021-03-04 21:39
 **/
@AllArgsConstructor
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.FILE_ISSUE_TOPIC, consumerGroup = MQConfig.FILE_ISSUE_CUSTOMERS_GROUP)
public class  IssueDelegate  implements RocketMQListener<String> {

    private ObjectMapper objectMapper;

    private IssueTaskMapper issueTaskMapper;

    private ConditionBuilder conditionBuilder;

    private IssueHandler issueHandler;

    private EsTimeMapper esTimeMapper;

    private FileExportMapper fileExportMapper;

    private IssueLookLikeHandler issueLookLikeHandler;

    private static final String LAST_CLOSING_DATE = "last_closing_date";

    @SneakyThrows
    @Override
    public void onMessage(String message) {
        long time = System.currentTimeMillis();
        IssueMqMsg issueMqMsg = objectMapper.readValue(message, IssueMqMsg.class);
        log.info("-------本地下发消费开始：{}",issueMqMsg.getId());

        try {
            Thread.sleep(1000);
            //幂等性控制，防止单次消费处理时间大于MQ默认15分钟触发重试机制
            int count = issueTaskMapper.transNum(issueMqMsg.getId());
            log.info("当前线程：{}，处理：{}，获取count:{}",Thread.currentThread().getName(),issueMqMsg.getId(),count);
            if (count > 0){
                log.error("任务：{}已经处理过",issueMqMsg.getId());
                return;
            }
            issueTaskMapper.addTransNum(issueMqMsg.getId());

            IssueTaskLog issueTaskLog = issueTaskMapper.findOneTaskLog(issueMqMsg.getId());
            if (StringUtils.isEmpty(issueTaskLog.getLookConfigId())){

                IssueTaskLog dbLog = issueTaskMapper.findIndexByCustomerId(issueTaskLog.getCustomersId());
                if (dbLog == null){
                    log.error("客群下发任务查询客群信息为空"+issueTaskLog.getCustomersId());
                    return;
                }
                issueTaskLog.setIndexId(dbLog.getIndexId());
                issueTaskLog.setTimeType(dbLog.getTimeType());

                issueTaskLog.setFields(issueTaskLog.getFields()+","+LAST_CLOSING_DATE);
                IssueLog issueLog = issueTaskMapper.findOneLog(issueMqMsg.getId());
                if (issueLog == null) {
                    log.error("客群下发任务执行日志为空"+issueMqMsg.getId());
                    return;
                }
                //此处根据dataDate去校验ES是否正在写入数据
                if (!checkWrite(issueLog.getId(),issueLog.getDataDate(),issueTaskLog.getIndexId())){
                    log.error("此时ES正在写入数据");
                    return;
                }
                List<Filter> filters = new ArrayList<>();
                FilterGroup filterGroup = conditionBuilder.findFilterGroup("-", issueTaskLog.getCustomersId(), filters)
                        .get(0);
                IssueResult issueResult = issueHandler.handler(issueTaskLog, issueLog, filterGroup);
                issueTaskMapper.update(new IssueLog().setId(issueMqMsg.getId()).setState("2").setReason("-")
                        .setUpdateTime(new Date()).setAmount(issueResult.getNum()).setPath(issueResult.getFilePath()));
                issueTaskMapper.deletePending(issueLog.getId());

            }else{
                //根据id获取导出配置信息
                LookLikeExportDto lookLikeConfig = fileExportMapper.findLookLikeConfig(issueTaskLog.getLookConfigId());
                IssueLog issueLog = issueTaskMapper.findOneLog(issueMqMsg.getId());
                IssueResult issueResult = issueLookLikeHandler.handler(issueTaskLog, issueLog, lookLikeConfig);
                issueTaskMapper.update(new IssueLog().setId(issueMqMsg.getId()).setState("2").setReason("-")
                        .setUpdateTime(new Date()).setAmount(issueResult.getNum()).setPath(issueResult.getFilePath()));
            }
            log.info("客群下发文件成功消费mq:{}",issueMqMsg.getId());
        } catch (Exception e) {
            log.error("******客群下发文件出现异常"+issueMqMsg.getId(), e);
            issueTaskMapper.update(new IssueLog().setId(issueMqMsg.getId()).setState("3").setReason("下发失败")
                    .setUpdateTime(new Date()));
            issueTaskMapper.deletePending(issueMqMsg.getId());
        } finally {
            log.debug("客群下发文件:{}总耗时=====>{}ms",issueMqMsg.getId(), (System.currentTimeMillis() - time));
        }
    }

    private boolean checkWrite(String logId, String dataDate,String indexId){
        boolean isCheck = true;
        if (!StringUtils.isEmpty(dataDate)){
            //说明是日更操作
            if (FieldHandler.pattern.matcher(dataDate).matches() && dataDate.length() == FieldHandler.LENGTH8){
                int count = issueTaskMapper.batchCount(1,dataDate,indexId);
                if (count>0){
                    issueTaskMapper.update(new IssueLog().setId(logId).setState("3").setReason("ES正在写入数据")
                            .setUpdateTime(new Date()));
                    isCheck = false;
                }
            }else{
                //对于下发来说，肯定是8位(立即下发或者周下发)或6位(月下发)日期，跟导出下载的校验不一样（8位日期或者后缀）。所以在
                //这里直接通过indexId找到上月末的后缀)
                String esSymbol = "";
                if (FieldHandler.INDEX_ONE.equals(indexId)){
                    esSymbol = Optional.ofNullable(esTimeMapper.findEsSymbol(indexId,FieldHandler.LAST_MONTH_ONE)).orElse(FieldHandler.LAST_MONTH_ONE);
                }else{
                    esSymbol = Optional.ofNullable(esTimeMapper.findEsSymbol(indexId,FieldHandler.LAST_MONTH_TWO)).orElse(FieldHandler.LAST_MONTH_TWO);
                }

                int count = issueTaskMapper.batchCount(2,esSymbol,indexId);
                if (count>0){
                    issueTaskMapper.update(new IssueLog().setId(logId).setReason("ES正在写入数据")
                            .setUpdateTime(new Date()));
                    isCheck = false;
                }

            }
        }
        return isCheck;
    }

}
